//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// update_executor.cpp
//
// Identification: src/execution/update_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//
#include <memory>

#include "execution/executors/update_executor.h"

namespace bustub {

UpdateExecutor::UpdateExecutor(ExecutorContext *exec_ctx, const UpdatePlanNode *plan,
                               std::unique_ptr<AbstractExecutor> &&child_executor)
    : AbstractExecutor(exec_ctx), plan_(plan), child_executor_(std::move(child_executor)) {
  // As of Fall 2022, you DON'T need to implement update executor to have perfect score in project 3 / project 4.
}

void UpdateExecutor::Init() {
  child_executor_->Init();
  table_oid_t table_id = plan_->TableOid();
  TableInfo *table = exec_ctx_->GetCatalog()->GetTable(table_id);
  table_info_ = table;
  indexs_ = exec_ctx_->GetCatalog()->GetTableIndexes(table->name_);
}

auto UpdateExecutor::Next([[maybe_unused]] Tuple *tuple, RID *rid) -> bool {
  if (table_info_ != nullptr) {
    bool updated = false;
    int updated_num = 0;
    while (child_executor_->Next(tuple, rid)) {
      updated = true;
      TupleMeta meta{};
      meta.delete_txn_id_ = meta.insert_txn_id_ = INVALID_TXN_ID;
      meta.is_deleted_ = true;
      table_info_->table_->UpdateTupleMeta(meta, *rid);
      meta.is_deleted_ = false;
      std::vector<bustub::Value> vals{};
      for (const auto &expr : plan_->target_expressions_) {
        vals.emplace_back(expr->Evaluate(tuple, table_info_->schema_));
      }
      Tuple inserted_tuple{vals, &table_info_->schema_};
      auto inserted_rid = table_info_->table_->InsertTuple(meta, inserted_tuple);
      if (inserted_rid == std::nullopt) {
        continue;
      }
      for (IndexInfo *index : indexs_) {
        const Tuple &key =
            inserted_tuple.KeyFromTuple(table_info_->schema_, index->key_schema_, index->index_->GetKeyAttrs());
        index->index_->InsertEntry(key, inserted_rid.value(), exec_ctx_->GetTransaction());
      }
      updated_num++;
    }
    if (once_) {
      updated = true;
      once_ = false;
    }
    if (updated) {
      WriteInsertedResult(updated_num, tuple);
    }
    return updated;
  }
  return false;
}

void UpdateExecutor::WriteInsertedResult(int inserted_num, Tuple *result) {
  std::vector<Column> columns{};
  columns.emplace_back(Column{"id", TypeId::INTEGER});
  Schema sch{columns};
  std::vector<bustub::Value> values{};
  values.emplace_back(Value{TypeId::INTEGER, inserted_num});
  *result = Tuple{values, &sch};
}

}  // namespace bustub
